package org.dromara.assets.service.impl;

import com.binance.api.client.BinanceApiClientFactory;
import com.binance.api.client.BinanceApiRestClient;
import com.binance.api.client.BinanceApiWebSocketClient;
import com.binance.api.client.exception.BinanceApiException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import java.io.Closeable;
import java.util.*;
import java.util.concurrent.*;

@Component
@Slf4j
@Scope("prototype") // 每次获取都是新实例
public class BinanceWebSocket {

    @Autowired
    BinanceWebSocketCallback binanceWebSocketCallback;

    BinanceApiRestClient client;
    BinanceApiWebSocketClient webSocketClient;
    String listenKey;
    Closeable userDataUpdateEventWebSocket;

    public void init(String apiKey, String apiSecret) {
        BinanceApiClientFactory factory = BinanceApiClientFactory.newInstance(apiKey, apiSecret);
        client = factory.newRestClient();
        webSocketClient = factory.newWebSocketClient();
        startWebSocket();
        // 每隔30分钟刷新Listen Key
        new Timer().schedule(new TimerTask() {
            @Override
            public void run() {
                keepAliveListenKey();
            }
        }, 0, 30 * 60 * 1000); // 30分钟
    }

    private final ScheduledExecutorService executorService =
            new ScheduledThreadPoolExecutor(1, new RejectedExecutionHandler() {
                @Override
                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                    log.error("websocket:任务被拒绝，因为前一个任务还未完成");
                }
            });

    private void scheduleReconnect() {
        executorService.schedule(() -> {
            try {
                startWebSocket();
            } catch (Exception e) {
                log.error("websocket:重连出现异常，将在5秒后再次尝试重连", e);
                // 重连异常，再次安排重连
                scheduleReconnect();
            }
        }, 5, TimeUnit.SECONDS);
    }

    private void startWebSocket() {
        try {
            log.info("websocket:开始订阅WebSocket消息");
            if (userDataUpdateEventWebSocket != null) {
                userDataUpdateEventWebSocket.close();
            }
            listenKey = getListenKey();
            // 开始接收订单更新消息
            subscribeOrderUpdates(webSocketClient, listenKey);
        } catch (Exception e) {
            log.error("", e);
            scheduleReconnect();
        }
    }

    private String getListenKey() {
        return client.startUserDataStream();
    }

    private void keepAliveListenKey() {
        try {
            client.keepAliveUserDataStream(listenKey);
            log.info("websocket:Listen Key 已刷新");
        } catch (BinanceApiException e) {
            log.error("websocket:刷新Listen Key失败: ", e);
            scheduleReconnect();
        }
    }

    private void subscribeOrderUpdates(BinanceApiWebSocketClient webSocketClient, String listenKey) {
        log.info("register callback event");
        userDataUpdateEventWebSocket = webSocketClient.onUserDataUpdateEvent(listenKey, binanceWebSocketCallback);
    }
}
